package com.ayu.shortlink.project.mq.consumer;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

/**
 * 延迟双删 Redis stream流消费者
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class DelayDeleteConsumer implements CommandLineRunner {
    private final RedissonClient redissonClient;
    private final StringRedisTemplate stringRedisTemplate;

    @Value("${short-link.longPolling}")
    private boolean longPolling;

    // 默认睡眠1s
    private Integer sleepTime = 1000;
    @Override
    public void run(String... args) throws Exception {
        Executors.newSingleThreadExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("DelayDeleteConsumer");
                thread.setDaemon(Boolean.TRUE);
                // 提供发生异常时的日志打印功能
                thread.setUncaughtExceptionHandler((t,e) -> {
                    log.error("DelayDeleteConsumer-thread:{}-exception:{}", t, e.getMessage());
                });
                return thread;
            }
        }).execute(() -> {
            RBlockingQueue<Object> delayDeleteQueue = redissonClient.getBlockingQueue("DELAY_DELETE_QUEUE");
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(delayDeleteQueue);
            // 如果开启长轮询 睡眠时间5s
            if (longPolling){
                sleepTime *= 5;
            }
            while (true){
                while(!delayedQueue.isEmpty()){
                    log.info("正在执行延迟双删");
                    stringRedisTemplate.delete((String) Objects.requireNonNull(delayedQueue.poll()));
                }

                // 暂时没有任务 可以暂时让出CPU时间片
                try {
                    Thread.sleep(sleepTime);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }
}
